package cn._51doit.day04;

import com.google.common.hash.*;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.nio.charset.Charset;


/**
 * @create: 2021-10-17 11:21
 * @author: 今晚打脑斧
 * @program: KeByDemo01
 * @Description:
 *  实时统计广告点击的人数和次数
 *  第二版,使用布隆过滤器
 * #广告id,用户id
 *    ad1,user1
 *    ad1,user1
 *    ad1,user2
 *    ad2,user1
 *    ad2,user2
 *
 *    #结果
 *    ad1,2,3
 *    ad2,2,2
 **/
public class ZuoYe2 {
    public static void main(String[] args) throws Exception {
        Configuration configuration = new Configuration();
        configuration.setInteger("rest.port", 8081);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);
        DataStreamSource<String> lines = env.socketTextStream("doit01", 8888);

        SingleOutputStreamOperator<Tuple3<String, String, Integer>> mapped = lines.map(new MapFunction<String, Tuple3<String, String, Integer>>() {
            @Override
            public Tuple3<String, String, Integer> map(String s) throws Exception {
                String[] split = s.split(",");
                return Tuple3.of(split[0], split[1], 1);
            }
        });
        //按照广告进行分区,相同广告肯定分到同一区内
        KeyedStream<Tuple3<String, String, Integer>, String> keyedStream = mapped.keyBy(tp -> tp.f0);

        SingleOutputStreamOperator<Tuple3<String, String, Integer>> reduce = keyedStream.reduce(new RichReduceFunction<Tuple3<String, String, Integer>>() {
            BloomFilter<CharSequence> bloomFilter;
            boolean flag ;
            int sum  ;

            //初始化布隆过滤器
            @Override
            public void open(Configuration parameters) throws Exception {
                bloomFilter = BloomFilter.create(Funnels.stringFunnel(Charset.forName("utf-8")), 10000, 0.01);
                flag = true;
                sum = 1;
            }

            //对数据进行滚动聚合,并且把人次加载进去
            @Override
            public Tuple3<String, String, Integer> reduce(Tuple3<String, String, Integer> in1, Tuple3<String, String, Integer> in2) throws Exception {
                sum +=  in2.f2;
                if (flag) {
                    bloomFilter.put(in1.f1);
                    flag=false;
                }
                if (!bloomFilter.mightContain(in2.f1)){
                bloomFilter.put(in2.f1);
                }
                long l = bloomFilter.approximateElementCount();
                return Tuple3.of(in1.f0, String.valueOf(l) , sum);
            }
        });
        //打印,把第一次的数据判断出来打印人次为1
        reduce.addSink(new SinkFunction<Tuple3<String, String, Integer>>() {
            boolean flagg = true ;
            @Override
            public void invoke(Tuple3<String, String, Integer> value) throws Exception {
                if (flagg){
                    System.out.println("广告 "+value.f0+" 人次 "+1+" 次数 "+value.f2);
                    flagg=false;
                }
                else
                    System.out.println("广告 "+value.f0+" 人次 "+value.f1+" 次数 "+value.f2);
            }
        });
        env.execute();
    }

}
